package com.atguigu.edu.realtime.utils;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.EduConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

public class HbaseUtil {


    //获取连接
    public static Connection getHbaseConnection(){

        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        conf.set("hbase.zookeeper.property.clientPort", "2181");


        try {
            Connection connection = ConnectionFactory.createConnection(conf);
            return connection;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //关闭连接
    public static void closeHbaseConnection(Connection connection){

        try {
            connection.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }
    //创建表
    public static void createHbaseTable(Connection connection , String nameSpace, String tableName, String... familes){
        if (familes ==null || familes.length < 1 ){
            System.out.println("需要指定列族");
            return;
        }

        try(Admin admin = connection.getAdmin()) {
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            if (admin.tableExists(tableNameObj)){
                System.out.println("你创建的"+nameSpace+"库的"+tableName+"表已经存在");
                return;
            }

            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);
            for (String famile : familes) {

                ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder =
                        ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(famile));
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
            }
            admin.createTable(tableDescriptorBuilder.build());

        } catch (IOException e) {
            throw new RuntimeException(e);
        }


    }
    //删除表
    public static void dropTable(Connection conn, String nameSpace, String tableName) {
        try (Admin admin = conn.getAdmin()) {
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            if (!admin.tableExists(tableNameObj)) {
                System.out.println("你要删除的" + nameSpace + "的" + tableName + "不存在");
                return;
            }
            System.out.println("删除" + nameSpace + "的" + tableName + "表");
            admin.disableTable(tableNameObj);
            admin.deleteTable(tableNameObj);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //插入数据
    public static void putRow(Connection connection ,String nameSpace ,String tableName ,
                              String rowKey,String columnFamily,String[] columnNames,String[] columnValues){
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);

        try(Table table = connection.getTable(tableNameObj)) {
            Put put = new Put(Bytes.toBytes(rowKey));
            for (int i = 0; i < columnNames.length ; i++) {
                String columnName = columnNames[i];
                String columnValue = columnValues[i];
                if (StringUtils.isNotEmpty(columnValue)){
                    put.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(columnName),Bytes.toBytes(columnValue));
                }
            }
            table.put(put);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //删除数据
    public static void delRow(Connection conn,String nameSpace,String tableName,String rowKey) throws IOException {
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        Table table = conn.getTable(tableNameObj);
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        table.delete(delete);
        table.close();
    }

    public static void main(String[] args) {
        Connection conn = getHbaseConnection();
        createHbaseTable(conn,"info","test","0");
        closeHbaseConnection(conn);
    }

    //获取支持异步操作的hbase连接
    public static AsyncConnection getAsyncHbaseConnection() {
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        try {
            AsyncConnection asyncConnection = ConnectionFactory.createAsyncConnection(conf).get();
            return asyncConnection;
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    //关闭支持异步操作的hbase连接
    public static void closeAsyncHbaseConnection(AsyncConnection asyncHbaseConn) {
        if (asyncHbaseConn != null && !asyncHbaseConn.isClosed()) {
            try {
                asyncHbaseConn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    //发送异步请求，根据rowkey从hbase表中查询数据
    public static JSONObject getDimInfoFromHbaseByAsync(AsyncConnection asyncHbaseConn,
                                                        String namespace,
                                                        String tableName,
                                                        String rowkey){
        try {
            TableName tableNameObj = TableName.valueOf(namespace, tableName);
            AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncHbaseConn.getTable(tableNameObj);
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = asyncTable.get(get).get();
            List<Cell> cells;
            cells = result.listCells();
            if(cells != null && cells.size() > 0){
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName,columnValue);
                }
                return jsonObj;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }


    //获取从hbase字典表中读取数据  创建动态表的建表语句
    public static String getBaseDicDDL(){
        return "CREATE TABLE test_paper (\n" +
                "id string,\n" +
                "info ROW<paper_title string,course_id string,create_time string,update_time string,publisher_id string,deleted string>,\n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") " + getHbaseDDL("dim_test_paper");
    }

    //获取hbase连接器连接属性
    public static String getHbaseDDL(String tableName){
        return " WITH (\n" +
                " 'connector' = 'hbase-2.2',\n" +
                " 'table-name' = '"+ "edu_realtime" +":"+tableName+"',\n" +
                " 'zookeeper.quorum' = 'hadoop102,hadoop103,hadoop104,hadoop105,hadoop106:2181',\n" +
                " 'lookup.async' = 'true',\n" +
                " 'lookup.cache' = 'PARTIAL',\n" +
                " 'lookup.partial-cache.max-rows' = '200',\n" +
                " 'lookup.partial-cache.expire-after-write' = '1 hour',\n" +
                " 'lookup.partial-cache.expire-after-access' = '1 hour'\n" +
                ")";
    }
}
